package test.zookeeper;

import java.io.IOException;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class Lock {
	
	private ZooKeeper zookeeper;
	private static ThreadLocal<String> tagLocal = new ThreadLocal<>();
	
	public Lock() {
		try {
			zookeeper = new ZooKeeper("127.0.0.1:2181", 3000, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					String path = event.getPath();
					EventType type = event.getType();
					KeeperState state = event.getState();
					System.out.println("path=" + path + ", type=" + type + ", state=" + state);
				}
			});
			Stat stat = zookeeper.exists("/lock", null);
			if(stat==null) {
				String result = zookeeper.create("/lock", null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				System.out.println(result);
			}
		} catch (IOException e) {
			e.printStackTrace();
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public void lock(String key) {
		String tag = tagLocal.get();
		key = "/lock/" + key;
		try {
			Stat stat = zookeeper.exists(key, null);
			if(stat==null) {
				zookeeper.create(key, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
			String result = zookeeper.create(key + "/index_", tag.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
			print(result);
			List<String> list = zookeeper.getChildren(key, false);
			Collections.sort(list);
			print(list.toString());
			if((key + "/" + list.get(0)).equals(result)) {
				print(new Date() + "---成功获得锁");
				return;
			}
			String nearIndex = null;
			for(String str : list) {
				if((key + "/" + str).equals(result)) {
					break;
				}
				nearIndex = str;
				System.out.println("nearIndex=" + str);
			}
			System.out.println("finally,nearIndex=" + nearIndex);
			CountDownLatch cdl = new CountDownLatch(1);
			zookeeper.exists(key + "/" + nearIndex, new Watcher() {
				
				@Override
				public void process(WatchedEvent event) {
					if(event.getType()==EventType.NodeDeleted) {
						cdl.countDown();
					}
				}
			});
			print(new Date() + "---坐等" + nearIndex + "被删除");
			cdl.await();
			print(new Date() + "---成功获得锁");
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public void lock(String key, int timeout) {
		String tag = tagLocal.get();
	}
	
	public boolean unlock(String key) {
		String tag = tagLocal.get();
		key = "/lock/" + key;
		try {
			List<String> list = zookeeper.getChildren(key, false);
			Collections.sort(list);
			if(CollectionUtils.isNotEmpty(list)) {
				String child = list.get(0);
				Stat stat = new Stat();
				byte[] bytes = zookeeper.getData(key + "/" + child, null, stat);
				if(tag.equals(new String(bytes))) {
					zookeeper.delete(key + "/" + child, stat.getVersion());
					return true;
				}
			}
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return false;
	}
	
	public void print(String str) {
		System.out.println(Thread.currentThread().getName() + "---" + str);
	}
	
	public static void main(String[] args) {
		new Thread() {
			@Override
			public void run() {
				tagLocal.set(Thread.currentThread().getName());
				Lock lock = new Lock();
				lock.lock("hello");
//				boolean isSuccess = lock.unlock("hello");
//				lock.print("解锁" + (isSuccess?"成功":"失败"));
			}
		}.start();
		
		new Thread() {
			@Override
			public void run() {
				tagLocal.set(Thread.currentThread().getName());
				try {
					Thread.sleep(5000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				Lock lock = new Lock();
				boolean isSuccess = lock.unlock("hello");
				lock.print("解锁" + (isSuccess?"成功":"失败"));
			}
		}.start();
		
	}

}
